package org.apache.flink.runtime.entrypoint;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.entrypoint.master.DefaultMasterComponentFactory;
import org.apache.flink.runtime.resourceconsumer.runner.ResourceConsumerServiceLeadershipRunnerFactory;
import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;

/**
 * @类名: StandaloneSessionClusterEntrypoint
 * @说明: Standalone集群入口点
 *
 * @author   leehom
 * @Date	 2023年12月27日 上午11:34:21
 * 修改记录：
 *
 * @see 	 
 */
/** Entry point for the standalone session cluster. */
public class StandaloneSessionClusterEntrypoint extends SessionClusterEntrypoint {

    public StandaloneSessionClusterEntrypoint(Configuration configuration) {
        super(configuration);
    }

    @Override
    protected DefaultMasterComponentFactory
            createMasterComponentFactory(Configuration configuration) {
    	// master组件工厂
        return DefaultMasterComponentFactory.createMasterComponentFactory(
        		ResourceConsumerServiceLeadershipRunnerFactory.INSTANCE,
                StandaloneResourceManagerFactory.getInstance());
    }

    public static void main(String[] args) {
        // startup checks and logging
        EnvironmentInformation.logEnvironmentInfo(
                LOG, StandaloneSessionClusterEntrypoint.class.getSimpleName(), args);
        SignalHandler.register(LOG);
        JvmShutdownSafeguard.installAsShutdownHook(LOG);
        // 抓取配置
        final EntrypointClusterConfiguration entrypointClusterConfiguration =
                ClusterEntrypointUtils.parseParametersOrExit(
                        args,
                        new EntrypointClusterConfigurationParserFactory(),
                        StandaloneSessionClusterEntrypoint.class);
        // 
        Configuration configuration = loadConfiguration(entrypointClusterConfiguration);
        // 构建入口点
        StandaloneSessionClusterEntrypoint entrypoint =
                	new StandaloneSessionClusterEntrypoint(configuration);
        // run
        ClusterEntrypoint.runClusterEntrypoint(entrypoint);
    }
}
